package com.analysis

//各国的购买与退货订单量关系

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._

object CountryPurchaseReturn {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .appName("Purchase vs. Return by Country")
      .master("local[*]")
      .getOrCreate()

    val cleanedDF = spark.read.parquet("output/cleaned_data.parquet")

    val purchasesDF = cleanedDF
      .filter(!col("InvoiceNo").startsWith("C"))
      .groupBy("Country")
      .agg(countDistinct("InvoiceNo").as("PurchaseCount"))

    val returnsDF = cleanedDF
      .filter(col("InvoiceNo").startsWith("C"))
      .groupBy("Country")
      .agg(countDistinct("InvoiceNo").as("ReturnCount"))

    val resultDF = purchasesDF
      .join(returnsDF, Seq("Country"), "full_outer")
      .na.fill(0, Seq("PurchaseCount", "ReturnCount"))
      .orderBy(desc("PurchaseCount"))

    resultDF.write
      .mode("overwrite")
      .json("output/analysis_results/country_purchase_return")

    resultDF.show()

    spark.stop()
  }
}